package cn.itcast.flink.source;

import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

/**
 * Author itcast
 * Date 2021/9/20 9:58
 * flink消费kafka集群的数据，并将其打印输出
 * 开发步骤：
 * 1.创建流执行环境
 * 2.设置相关参数
 * 3.创建FlinkKafkaConsumer，用于读取kafka消费者
 * 4.设置相关的配置
 * 5.将消费的数据添加到数据源
 * 6.打印输出
 * 7.执行流环境
 */
public class FlinkKafkaReader {
    public static void main(String[] args) throws Exception {
        //1.创建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.设置相关参数
        env.setParallelism(1);
        //4.设置相关的配置
        Properties props = new Properties();
        //4.1.连接kafka集群的地址
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node01:9092,node02:9092,node03:9092");
        //4.2.消费者组
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"_vehicle_data_consumer");
        //3.创建FlinkKafkaConsumer，用于读取kafka消费者
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
                "vehicledata",
                new SimpleStringSchema(),
                props
        );
        //设置consumer参数
        consumer.setStartFromEarliest();

        //5.将消费的数据添加到数据源
        DataStreamSource<String> source = env.addSource(consumer);
        SingleOutputStreamOperator<String> source1 = source.map(new RichMapFunction<String, String>() {
            LongCounter count = new LongCounter();

            @Override
            public void open(Configuration parameters) throws Exception {
                getRuntimeContext().addAccumulator("count-kafka", count);
            }

            @Override
            public String map(String value) throws Exception {
                count.add(1);
                return value;
            }
        });
        //6.打印输出
        source1.printToErr();
        //7.执行流环境
        JobExecutionResult execute = env.execute();
        //
        Object accumulatorResult = execute.getAccumulatorResult("count-kafka");
        System.out.println(accumulatorResult);
    }
}
